package net.bwie.flink;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * 	2. FlinkSQL 开发
 * 		编写SQL语句
 * 			//1-表执行环境-tabEnv
 * 			//2-输入表-inpuTable
 * 			//3-查询-select
 * 			//4-输出表-outputTable
 * 			//5-插入-insert
 * @author xuanyu
 * @date 2025/10/21
 */
public class _04FlinkSqlDemo {

	public static void main(String[] args) {
		//1-表执行环境-tabEnv
		/*
		https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/tableapi/#overview--examples
		 */
		EnvironmentSettings settings = EnvironmentSettings
			.newInstance()
			.inStreamingMode()
			.build();
		TableEnvironment tabEnv = TableEnvironment.create(settings);

		//2-输入表-inputTable
/*
Kafka 队列：order-topic
数据：
u1,water,2025-10-21 14:44:23,5.0
u1,milk,2025-10-21 14:45:23,8.0
u1,bread,2025-10-21 14:45:29,6.0
u1,smoke,2025-10-21 14:46:23,8.0

控制台生成者发送数据
/opt/module/kafka/bin/kafka-console-producer.sh --broker-list node101:9092,node102:9092,node103:9092 --topic order-topic

*/
		tabEnv.executeSql(
		"CREATE TABLE tbl_order_kafka_source (\n" +
			"  `user_name` STRING,\n" +
			"  `item_name` STRING,\n" +
			"  `order_time` STRING,\n" +
			"  `order_amount` DECIMAL(10, 2)\n" +
			") WITH (\n" +
			"  'connector' = 'kafka',\n" +
			"  'topic' = 'order-topic',\n" +
			"  'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
			"  'properties.group.id' = 'testGroup',\n" +
			"  'scan.startup.mode' = 'earliest-offset',\n" +
			"  'format' = 'csv'\n" +
			")"
		);
		//tabEnv.sqlQuery("SELECT * FROM tbl_order_kafka_source").execute().print();

		//3-查询-select
		/*
			todo 实时统计总订单数和总销售额
		 */
		Table table = tabEnv.sqlQuery(
			"SELECT \n" +
				"    count(item_name) AS total_order_count\n" +
				"    , sum(order_amount) AS total_order_amount\n" +
				"FROM tbl_order_kafka_source"
		);
		// table.execute().print();
		tabEnv.createTemporaryView("result_table", table);

		//4-输出表-outputTable
		tabEnv.executeSql(
			"CREATE TABLE order_report_mysql_sink (\n" +
				"    total_order_count BIGINT,\n" +
				"    total_order_amount DECIMAL(10, 2),\n" +
				"    PRIMARY KEY (total_order_count) NOT ENFORCED\n" +
				") WITH (\n" +
				"   'connector' = 'jdbc',\n" +
				"   'url' = 'jdbc:mysql://node101:3306/gmall_report',\n" +
				"   'table-name' = 'order_report',\n" +
				"   'username' = 'root',\n" +
				"   'password' = '123456',\n" +
				"   'driver' = 'com.mysql.jdbc.Driver'\n" +
				")"
		);

		//5-插入-insert
		tabEnv.executeSql(
			"INSERT INTO order_report_mysql_sink\n" +
				"SELECT total_order_count, total_order_amount FROM result_table"
		);

	}

}
